package com.lagou.no1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object StructuredMysql {
    def main(args: Array[String]): Unit = {
        //设置控制台日志级别
        Logger.getLogger("org").setLevel(Level.WARN)
        //获取sparkSession对象
        val spark: SparkSession = SparkSession.builder()
          .master("local[*]")
          .appName(this.getClass.getCanonicalName)
          .getOrCreate()
        import spark.implicits._
        //连接kafka，获取流数据
        val kafkaDF: DataFrame = spark.readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "node01:9092")
          .option("subscribe", "lg_bus_info")
          .load()
        //获取value中的数据
        val kafkaValueDF: DataFrame = kafkaDF.selectExpr("cast(value as string)")
        //转为dataSet
        val kafkaDS: Dataset[String] = kafkaValueDF.as[String]
        //将ds中的字符串转为样例类，方便后续操作
        val busInfoDS: Dataset[BusInfo] = kafkaDS.map(str => BusInfo(str))
        //输出到mysql
        val write = new MysqlWritte
        busInfoDS.writeStream
          .foreach(write)
          .outputMode("append")
          .option("checkpointLocation","file:///F:\\作业\\formal-stage5-module1\\code\\homework\\data\\ckp\\")
          .start()
          .awaitTermination()

    }
}
